<?php
/**
 * Message Queue of Events
 */
class EventsMQ
{
	private static $_instance_default = NULL;
	private static $_instance_cpc     = NULL;
	private $_mq;

	public static function handle ($event, & $param)
	{
		$msg = NULL;
		switch ($event) 
		{
			case Kohana::config('events.user.create'):
				$msg = $param;
				$msg['type'] = 'passport';
				break;

			case Kohana::config('events.im.click'):
				$msg = $param;
				$msg['type'] = 'my_im';
				break;

			case Kohana::config('events.system.mass-mail'):
				$msg = $param;
				$msg['type'] = 'sys_massmail';
				break;

			case Kohana::config('events.system.mass-sms'):
				$msg = $param;
				$msg['type'] = 'sys_masssms';
				break;

			case Kohana::config('events.shop.type_changed'):
				$msg = $param;
				$msg['type'] = 'stats_shop';
				break;

			case Kohana::config('events.cuxiao.created'):
				$msg = $param;
				$msg['type'] = 'stats_cuxiao';
				break;

			case Kohana::config('events.cuxiao.refreshed'):
				$msg = $param;
				$msg['type'] = 'stats_cuxiao';
				break;

			case Kohana::config('events.trade.pay_order_success'):
				$msg = $param;
				$msg['type'] = 'stats_order';
				$msg['order_type'] = 'paid';
				$msg['day']  = date('Y-m-d');
				break;

			case Kohana::config('events.loupan.bangding'):
				$msg = $param;
				$msg['type'] = 'stats_loupan';
				$time = Arr::get($param,'time');
				$msg['day']  = date('Y-m-d',$time);
				break;
				
			case Kohana::config('events.cuxiao.resize_img'):
				$msg = $param;
				$msg['type'] = 'resize_img';
				break;

			case Kohana::config('events.shop.first_access'):
			case Kohana::config('events.shop.new_cuxiao'):
				$msg = $param;
				$msg['type'] = 'stats_shopbehavior';
				break;

			default:
				return;
				break;
		}

		if (NULL !== $msg)
		{
			if ($msg['type'] == 'resize_img')
			{
				$result = self::instance()->add($msg,'img');
			}
			else
			{
				self::instance()->add($msg);
			}
		}
	}

	public static function instance ($group = 'default')
	{
		//
		// 暂不实现$group
		//
		if ($group == 'default')
        {
		    if (self::$_instance_default === NULL)
		    {
			    $class = __CLASS__;
			    self::$_instance_default = new $class('redis');
		    }

		    return self::$_instance_default;
        }
        else if ($group == 'CPC')
        {
		    if (self::$_instance_cpc === NULL)
		    {
			    $class = __CLASS__;
			    self::$_instance_cpc = new $class('cpc');
		    }

		    return self::$_instance_cpc;
        }
        else
        {
			throw new Exception("not support group '$group'");
        }
	}

	/**
	 * @param	array	$msgValue
	 * @param	string	$msgType
	 * @return	boolean
	 */
	public function add ($msgValue, $key, $msgType = 'default')
	{
		return $this->_mq->lPush($key, $msgValue);
	}

	/**
	 * @param	string	$msgType
	 * @return	boolean
	 */
	public function get ($key = 'sdkcmds', $msgType = 'default')
	{
		$v   = $this->_mq->rPop($key);
        return $v;
	}

	public function listen ($group = 'default')
	{
		// only run at cli mode
		if (PHP_SAPI !== 'cli')
			throw new Exception("only run at cli mode");

		while (TRUE)
		{
			$pid  = pcntl_fork();
			if ($pid < 0)
			{
				die ("Cannot fork child process");
			}
			else if ($pid > 0) 
			// parent
			{
				$status = 0;
				pcntl_waitpid($pid, $status);
			}
			else	
			// child
			{
				$msgs_count = Kohana::config('eventsmq.handler.messages_per_child');
				$handler   = new EventsMQ_Handler;
				$handler->handle_event($msgs_count,$group);
				exit;
			}

		} // while
	}

	protected function __construct ($group_name = 'redis')
	{
		$this->_mq = new Redis;
		$this->_mq->pconnect(Kohana::config("eventsmq.{$group_name}.host"), Kohana::config("eventsmq.{$group_name}.port"));
	}
}
